package reactive;

import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.scheduler.Schedulers;

public class T2 {
    private static Logger log = LoggerFactory.getLogger(T2.class);

    public static void main(String[] args) {
        Flux.create(fluxSink -> {
            int i = 0;
            while (i < 10000000) {
                fluxSink.next(i);
                i++;
            }
            fluxSink.complete();
        }, FluxSink.OverflowStrategy.LATEST).parallel(4).runOn(Schedulers.newParallel("T")).subscribe(new BaseSubscriber<Object>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                subscription.request(1);
            }

            @Override
            protected void hookOnNext(Object value) {
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info("receive :" + value);
                request(1);
            }
        });
        System.out.println(1);
    }
}
